package com.millstein.realtime.app.dwd.db;

import com.millstein.realtime.app.base.BaseSqlApp;
import com.millstein.realtime.common.Constants;
import com.millstein.realtime.util.SqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @Description
 * @Author tsing
 * @Date 2024-10-11 15:25
 */
public class Dwd_08_TradePayDetailSuc extends BaseSqlApp {

    public static void main(String[] args) {
        new Dwd_08_TradePayDetailSuc().init(
                7002,
                3,
                "Dwd_08_TradePayDetailSuc"
        );
    }

    /**
     * 具体数据处理的逻辑，由子类编写
     *
     * @param env      执行环境
     * @param tableEnv 表执行环境
     */
    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) {
        // 1.设置数据过期时间。假设下单到支付的有效时间是30分钟
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(30 * 60 + 5));

        // 2.从下单详情主题中获取数据
        tableEnv.executeSql(
                "create table order_detail (  " +
                "    id string,  " +
                "    order_id string,  " +
                "    sku_id string,  " +
                "    sku_name string,  " +
                "    create_time string,  " +
                "    source_id string,  " +
                "    source_type_code string,  " +
                "    source_type_name string,  " +
                "    sku_num string,  " +
                "    split_origin_amount string,  " +
                "    split_total_amount string,  " +
                "    split_activity_amount string,  " +
                "    split_coupon_amount string,  " +
                "    od_ts bigint,  " +
                "    user_id string,  " +
                "    province_id string,  " +
                "    operate_time string,  " +
                "    order_status string,  " +
                "    oi_ts bigint,  " +
                "    activity_id string,  " +
                "    activity_rule_id string,  " +
                "    coupon_id string,  " +
                "    row_opt_ts timestamp_ltz(3)  " +
                ")" + SqlUtil.getKafkaSourceDDL(Constants.TOPIC_DWD_TRADE_ORDER_DETAIL, "Dwd_08_TradePayDetailSuc")
        );

        // 3.从ods层中读取数据
        readOdsDataFromKafka(tableEnv, "Dwd_08_TradePayDetailSuc");

        // 4.从ods层数据中筛选出支付成功的数据
        Table paymentInfo = tableEnv.sqlQuery(
                "select  " +
                "    `data`['user_id'] user_id,  " +
                "    `data`['order_id'] order_id,  " +
                "    `data`['payment_type'] payment_type,  " +
                "    `data`['callback_time'] callback_time,  " +
                "    `pt`,  " +
                "    `ts`  " +
                "from maxwell_table  " +
                "where `database` = 'gmall'  " +
                "    and `table` = 'payment_info'  " +
                "    and `type` = 'update'  " +
                "    and `old`['payment_status'] is not null  " +
                "    and `data`['payment_status'] = '1602'"
        );
        tableEnv.createTemporaryView("payment_info", paymentInfo);

        // 5.读取base_dic字典表中的数据
        readBaseDicFromMysql(tableEnv);

        // 6.三张表join
        Table resultTable = tableEnv.sqlQuery(
                "select  " +
                "      od.id order_detail_id,  " +
                "      od.order_id,  " +
                "      od.user_id,  " +
                "      od.sku_id,  " +
                "      od.sku_name,  " +
                "      od.province_id,  " +
                "      od.activity_id,  " +
                "      od.activity_rule_id,  " +
                "      od.coupon_id,  " +
                "      pi.payment_type payment_type_code,  " +
                "      bd.dic_name payment_type_name,  " +
                "      pi.callback_time,  " +
                "      od.source_id,  " +
                "      od.source_type_code,  " +
                "      od.source_type_name,  " +
                "      od.sku_num,  " +
                "      od.split_origin_amount,  " +
                "      od.split_activity_amount,  " +
                "      od.split_coupon_amount,  " +
                "      od.split_total_amount split_payment_amount,  " +
                "      pi.ts,  " +
                "      od.row_opt_ts row_opt_ts  " +
                "from order_detail od  " +
                "join payment_info pi on pi.order_id = od.order_id  " +
                "join base_dic for system_time as of pi.pt as bd on pi.payment_type = bd.dic_code"
        );

        // 7.创建kafka动态表
        tableEnv.executeSql(
                "create table dwd_trade_pay_detail_suc (  " +
                "    order_detail_id string,  " +
                "    order_id string,  " +
                "    user_id string,  " +
                "    sku_id string,  " +
                "    sku_name string,  " +
                "    province_id string,  " +
                "    activity_id string,  " +
                "    activity_rule_id string,  " +
                "    coupon_id string,  " +
                "    payment_type_code string,  " +
                "    payment_type_name string,  " +
                "    callback_time string,  " +
                "    source_id string,  " +
                "    source_type_code string,  " +
                "    source_type_name string,  " +
                "    sku_num string,  " +
                "    split_origin_amount string,  " +
                "    split_activity_amount string,  " +
                "    split_coupon_amount string,  " +
                "    split_payment_amount string,  " +
                "    ts bigint,  " +
                "    row_opt_ts timestamp_ltz(3)  " +
                ")" + SqlUtil.getKafkaSinkDDL(Constants.TOPIC_DWD_TRADE_PAY_DETAIL_SUC)
        );
        
        // 8.将join后的数据写入动态表
        resultTable.executeInsert("dwd_trade_pay_detail_suc");
    }
}
